package com.wjw.redis.stream.service;

import com.alibaba.fastjson.JSONObject;
import com.wjw.redis.stream.constant.CacheConstant;
import com.wjw.redis.stream.model.UserInfo;
import com.wjw.redis.stream.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.connection.stream.StreamRecords;
import org.springframework.data.redis.connection.stream.StringRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.time.Instant;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import java.time.LocalDateTime;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author ：wjw
 * &#064;date ：Created in 2022/5/19 9:29
 * &#064;description：生产者
 */
@Slf4j
@Service
public class ProducerServiceImpl {

    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private RedisTemplate redisTemplate;
//    @Autowired
//    private TestAsync testAsync;

    private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); // 时间格式化器

    // 使用ConcurrentHashMap以支持线程安全
    private static final Map<String, String> sharedMap = new ConcurrentHashMap<>();


    private static final AtomicLong sequence = new AtomicLong(0);

    /**
     * create by: wjw
     * description: 每100秒生产10条数据，放入stream中
     * create time: 2022/5/19 9:39
     */
    @Scheduled(fixedDelay = 100000)
    public void sendRecord() {
        String streamKey = CacheConstant.REDIS_STEAM_02;//stream key
        for (int i = 0; i < 10; i++) {
            try {
                long time = System.currentTimeMillis();
                long seq = sequence.incrementAndGet();
                LocalDateTime dateTime = LocalDateTime.ofInstant(Instant.ofEpochSecond(time / 1000), ZoneOffset.ofHours(8));
                Map<String, String> map = new ConcurrentHashMap<>();
                map.put("a", "aa" + seq);
                map.put("b", "bb" + seq);
                map.put("c", "cc" + seq);
                map.put("time", dateTime.format(formatter)); // 格式化时间戳

                String result = redisUtil.addMap(CacheConstant.REDIS_STEAM_01, map);
                log.info("发送返回结果：{}", result);

//            //record新增的几种方式
//            UserInfo userInfo = new UserInfo("张三" + i, "男");
//            Record record = ObjectRecord.create(streamKey, userInfo).withId(RecordId.autoGenerate());
////            Record record = Record.of(userInfo).withStreamKey(streamKey);
////            Record record = Record.of(userInfo).withStreamKey(streamKey).withId(RecordId.autoGenerate());//指定id
////            Record record = StreamRecords.objectBacked(userInfo).withStreamKey(streamKey);
////            Record record =   StreamRecords.mapBacked(map).withStreamKey(streamKey).withId(RecordId.autoGenerate());//指定id
//            String recordId = redisUtil.addRecord(record); //返回的RecordId
//            log.info("发送返回结果：{}", recordId);
            } catch (Exception e) {
                log.error("发送数据到Stream中发生异常：", e); // 捕获并记录异常信息
            }
        }
    }

}
